package com.hunter.tc.kafka.serialization.avrolocal;

import org.apache.avro.file.DataFileWriter;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecordBase;
import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Map;

public class AvroLcoalSerializer<T> implements Serializer<T> {
    private DatumWriter<T> datumWriter;
    private DataFileWriter<T> dataFileWriter;
    private Class<T> tClass;
    private ByteArrayOutputStream byteArrayOutputStream;


    //required by the kafka
    public AvroLcoalSerializer() {
    }

    @Override
    public void configure(Map<String, ?> props, boolean isKey) {
        tClass = (Class<T>) props.get(AvroLocalSerdeConfig.avroPOJOClassName);
        if (tClass == null) {
            throw new IllegalArgumentException("need to set AvroLocalSerdeConfig.avroPOJOClassName");
        }
        //if(SpecificRecordBase.class.isAssignableFrom(tClass)){
        if (tClass.isAssignableFrom(SpecificRecordBase.class)) {
            throw new IllegalArgumentException("avroPOJOClassName needs to be a subclass of the SepcificRecordBase");
        }
        datumWriter = new SpecificDatumWriter<T>(tClass);
        dataFileWriter = new DataFileWriter<T>(datumWriter);
        byteArrayOutputStream = new ByteArrayOutputStream();
    }

    @Override
    public byte[] serialize(String topic, T data) {
        SpecificRecordBase recordBase = (SpecificRecordBase) data;
        try {
            dataFileWriter.create(recordBase.getSchema(), byteArrayOutputStream);
            return byteArrayOutputStream.toByteArray();
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public void close() {

    }
}
